"""
author: liuxu
date: 2025/5/1
description: hsa-trace测试
"""
#TODO http://10.0.54.20/dcutoolkit/driverruntime/hsa-class抽空测一下这个测试集看能跑起来不

import allure
import pytest
import re
from common.logger_ctrl import mylogger
from common.ssh_command import exec_cmd
from base_public.testbase import BasePublic

@allure.feature('hsa tracing指令')
@pytest.mark.medium
class TestHsaTrace(BasePublic):
    """测试--hsa-trace功能"""
    @pytest.mark.smoke
    @pytest.mark.high
    def test_hsa_trace(self, test_env):
        """
        测试跟踪hsa接口
        :param test_env: 
        :return: 
        """
        testcase = self.compile_source_code(src_file_name="hipSimple/matrix.cpp")
        ret, output = exec_cmd(f"hipprof --hsa-trace {testcase}", cwd=test_env)
        assert ret is True
        # 获取生成的hsa的csv文件，读取csv的trace数据
        hsa_csv_path = self.get_all_prof_files_info(output, test_env)["hsa_api_csv"]["abspath"]
        hsa_trace_d = self.get_trace_info_from_csv(hsa_csv_path)
        # 断言HIP PROF:hsa API statistics有生成hsa结果，且程序总运行时间小于1s(10的9次方纳秒)
        assert len(hsa_trace_d) > 10 and hsa_trace_d["Total"]["TotalDurationNs"] < 1 * (10**9)

    def test_hsa_multi_process(self, test_env):
        """
        测试多卡多进程多线程用例
        :param test_env: 
        :return: 
        """
        testcase = self.compile_source_code(src_file_name="multiproc_case/multiprocess_multithread_multidevice_test.cpp")
        ret, output = exec_cmd(f"hipprof --hsa-trace {testcase}", cwd=test_env)
        assert ret is True
        # 获取生成的libc的csv文件，读取csv的trace数据
        hsa_csv_path = self.get_all_prof_files_info(output, test_env)["hsa_api_csv"]["abspath"]
        hsa_trace_d = self.get_trace_info_from_csv(hsa_csv_path)
        assert len(hsa_trace_d) > 0

    def test_hsa_hid_args(self, test_env):
        """
        测试不收集函数参数
        :param test_env: 
        :return: 
        """
        testcase = self.compile_source_code(src_file_name="hipSimple/matrix.cpp")
        # ret, output = exec_cmd(f"hipprof --hsa-trace --hid-args {testcase}", cwd=test_env)
        ret, output = exec_cmd(f"hipprof --hsa-trace {testcase}", cwd=test_env)
        assert ret is True
        # 获取生成的libc的csv文件，读取csv的trace数据
        hsa_csv_path = self.get_all_prof_files_info(output, test_env)["hsa_api_csv"]["abspath"]
        hsa_trace_d = self.get_trace_info_from_csv(hsa_csv_path)
        for info_dict in hsa_trace_d:
            if "name" in info_dict and "args" in info_dict:
                assert info_dict["args"].get("args", "") == "", f'{info_dict["name"]}函数的参数不为空'

    def test_hsa_buffer_size(self, test_env):
        """
        设置数据刷新频率
        :param test_env: 
        :return: 
        """
        testcase = self.compile_source_code(src_file_name="hipSimple/matrix.cpp")
        ret, output = exec_cmd(f"hipprof --hsa-trace --buffer-size 1 {testcase}", cwd=test_env)
        assert ret is True
        # 获取生成的libc的csv文件，读取csv的trace数据
        hsa_csv_path = self.get_all_prof_files_info(output, test_env)["hsa_api_csv"]["abspath"]
        hsa_trace_d = self.get_trace_info_from_csv(hsa_csv_path)
        # 断言HIP PROF:libc API statistics有生成libc结果，且程序总运行时间小于1s(10的9次方纳秒)
        assert len(hsa_trace_d) > 0 and hsa_trace_d["Total"]["TotalDurationNs"] < 1 * (10 ** 9)

    def test_hsa_index_range(self, test_env):
        """
        按执行顺序过滤api数据
        :param test_env: 
        :return: 
        """
        testcase = self.compile_source_code(src_file_name="hipSimple/matrix.cpp")
        ret, output = exec_cmd(f"hipprof --hsa-trace --hip-trace --index-range 2:3 {testcase}", cwd=test_env)
        assert ret is True
        # 获取生成的libc的csv文件，读取csv的trace数据
        hsa_csv_path = self.get_all_prof_files_info(output, test_env)["hsa_api_csv"]["abspath"]
        hsa_trace_d = self.get_trace_info_from_csv(hsa_csv_path)
        assert len(hsa_trace_d) > 0

    def test_hsa_omp_case(self, test_env):
        """
        测试omp程序
        :param test_env: 
        :return: 
        """
        testcase = self.compile_omp_case()
        ret, output = exec_cmd(f"hipprof --hsa-trace {testcase}", cwd=test_env)
        assert ret is True
        # 获取生成的libc的csv文件，读取csv的trace数据
        hsa_csv_path = self.get_all_prof_files_info(output, test_env)["hsa_api_csv"]["abspath"]
        hsa_trace_d = self.get_trace_info_from_csv(hsa_csv_path)
        assert len(hsa_trace_d) > 0

    def test_hsa_python(self, test_env):
        """
        测试未调用hsa接口的程序
        :param test_env: 
        :return: 
        """
        ret, output = exec_cmd(f"hipprof --hsa-trace python3 -c 'print(1+1)'", cwd=test_env)
        assert (ret is True and "HIP_PROF:No hip trace info" in output) or ret is False

    def test_hsa_roctracer(self, test_env):
        """
        测试roctracer程序的hsa接口
        :param test_env:
        :return:  
        """
        exe_path = self.compile_source_code(src_file_name='roctracer_case/roctracer_test.cpp',
                                            compile_paras=f"-I {self.rocm_path}/roctracer/include -O0 -g")

        ret, output = exec_cmd(f"hipprof --hsa-trace {exe_path}", cwd=test_env)
        assert ret is True
        # 获取生成的libc的csv文件，读取csv的trace数据
        hsa_csv_path = self.get_all_prof_files_info(output, test_env)["hsa_api_csv"]["abspath"]
        hsa_trace_d = self.get_trace_info_from_csv(hsa_csv_path)
        assert len(hsa_trace_d) > 0